package com.atguigu.qzpoint.streaming

import java.lang
import java.sql.ResultSet
import java.util.Random

import com.atguigu.qzpoint.util.{DataSourceUtil, QueryCallback, SqlProxy}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

object RegisterStreaming {
  private val groupid = "register_group_test"

  def main(args: Array[String]): Unit = {
    //把程序写到HDFS中，需要配置这个，涉及到权限问题 超级管理员就是hdfs，换成这个就没有权限问题了
    System.setProperty("HADOOP_USER_NAME", "atguigu")

    //spark参数，构建spark环境
    val conf = new SparkConf().setAppName(this.getClass.getSimpleName)
     //最大接收条数 kafka
      .set("spark.streaming.kafka.maxRatePerPartition", "100")
      //背压
//      .set("spark.streaming.backpressure.enabled", "true")
      //优雅关闭
//      .set("spark.streaming.stopGracefullyOnShutdown", "true")
      .setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(3))
    val sparkContext: SparkContext = ssc.sparkContext


    // topic
    val topics = Array("register_topic")
    //消费kafka 用map来封装参数
    val kafkaMap: Map[String, Object] = Map[String, Object](
      //ip端口
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      //反序列化
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      //消费者组 共享offset
      "group.id" -> groupid,
      //第一次启动从哪里读
        // earliest  从最小的开始读
        // latest 从程序执行之后，发过来的数据才读
      "auto.offset.reset" -> "earliest",   //sparkstreaming第一次启动，不丢数
      //如果是true，则这个消费者的偏移量会在后台自动提交，但是kafka宕机容易丢失数据
      //如果是false，则需要手动维护kafka偏移量
      "enable.auto.commit" -> (false: lang.Boolean)
    )

    //配置hadoop 在虚拟机上执行  从core-site.xml、hdfs-site.xml、hive-site.xml中找配置信息
//    sparkContext.hadoopConfiguration.set("fs.defaultFS", "hdfs://nameservice1")
//    sparkContext.hadoopConfiguration.set("dfs.nameservices", "nameservice1")

    //sparkStreaming对有状态的数据操作，需要设定检查点目录，然后将状态保存到检查点中
    //spark检查点 原理和作用 与Flink的对比
    ssc.checkpoint("/user/atguigu/sparkstreaming/checkpoint")

    //查询mysql中是否有偏移量
    val sqlProxy = new SqlProxy()
    //公用offsetmap
    val offsetMap = new mutable.HashMap[TopicPartition, Long]()
    val client = DataSourceUtil.getConnection
    //用回调函数比较巧妙，减少代码量
    try {
      //sqlProxy用来拼装sql
      sqlProxy.executeQuery(client, "select * from `offset_manager` where groupid=?", Array(groupid), new QueryCallback {
        override def process(rs: ResultSet): Unit = {
          while (rs.next()) {
            val model = new TopicPartition(rs.getString(2), rs.getInt(3))
            val offset = rs.getLong(4)
            offsetMap.put(model, offset)
          }
          rs.close() //关闭游标
        }
      })
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      sqlProxy.shutdown(client)
    }

    //获得数据流
    //设置kafka消费数据的参数  判断本地是否有偏移量  有则根据偏移量继续消费 无则重新消费
    val stream: InputDStream[ConsumerRecord[String, String]] = if (offsetMap.isEmpty) {
      //直连方式的流（消费kafka就用这个，其他的都摒弃了）
      //第一次消费（次：指的是程序初始化的次数，第二次即第一次挂了重启程序则为第二次）
      KafkaUtils.createDirectStream(
        ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaMap))
    } else {
      //非第一次消费
      KafkaUtils.createDirectStream(
        ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaMap, offsetMap))
    }

    /*
    CREATE TABLE `offset_manager` (
    `groupid` varchar(50) DEFAULT NULL,
    `topic` varchar(50) DEFAULT NULL,
    `partition` int(11) DEFAULT NULL,
    `untiloffset` mediumtext,
    UNIQUE KEY `offset_unique` (`groupid`,`topic`,`partition`)
    ) ENGINE=InnoDB DEFAULT CHARSET=latin1;
     */
    //stream原始流无法进行使用和打印，会报序列化错误，所以需要做下面的map转换

    val resultDStream: DStream[(String, Int)] = stream
      //实际生产是不会用分隔符来切割数据的
      //大部分json，小部分用/001,/002这样打不出来的分割（其实这样数据量也会大）。
      // json会造成算数据冗余，是前后端协商的结果，他们更好解析。百万日活，数据量也不是特别大，考虑到安全性、稳定性。

      //有可能存在错行的情况
      .filter(item => item.value().split("\t").length == 3)
      //逻辑计算
      .mapPartitions(partitions => {
        partitions.map(item => {
          //取出内容 （key：offset）
          val line = item.value()
          val arr = line.split("\t")
          //匹配第二个元素
          val app_name = arr(1) match {
              //不太好，数据量变大了，一个字符比较好。如果数据只有1、2、3其实不变换更好
            case "1" => "PC"
            case "2" => "APP"
            case _ => "Other"
          }
          //返回
          (app_name, 1)
        })
      })
    //缓存  当一个流要用多次时，避免被释放掉
    resultDStream.cache()
    //(PC,1),(PC,1),(APP,1),(Other,1),(APP,1),(Other,1),(PC,1),(APP,1)
    //"=================每6s间隔1分钟内的注册数据================="
    //窗口间隔 6s  窗口长度 1min   计算范围：只有数据在窗口内才统计
    resultDStream.reduceByKeyAndWindow((x: Int, y: Int) => x + y, Seconds(60), Seconds(6)).print()
    //"========================================================="

    //"+++++++++++++++++++++++实时注册人数+++++++++++++++++++++++"//状态计算
    val updateFunc =
    //函数 values：分组后的序列  state：状态
      (values: Seq[Int], state: Option[Int]) => {
      val currentCount = values.sum //本批次求和
      val previousCount = state.getOrElse(0) //历史数据
        //返回 本批次+历史数据
      Some(currentCount + previousCount)
    }
    //updateStateByKey(函数) = 结果
    resultDStream.updateStateByKey(updateFunc).print()
    //"++++++++++++++++++++++++++++++++++++++++++++++++++++++++"

    //数据倾斜的解决方案
    val dsStream: DStream[(String, Int)] = stream.filter(item => item.value().split("\t").length == 3)
      .mapPartitions(partitions =>
        partitions.map(item => {
          val rand = new Random()
          val line = item.value()
          val arr = line.split("\t")
          val app_id = arr(1)
          //把数据打散 随机数_appid  不要判断id，不要打散太多，不划算
          (rand.nextInt(3) + "_" + app_id, 1)
        }))

    //现在聚合是打散之后的结果
    val result: DStream[(String, Int)] = dsStream.reduceByKey(_ + _)

    //把id还原之后再聚合一次
    result.map(item => {
      val appid = item._1.split("_")(1)
      (appid, item._2)
    }).reduceByKey(_ + _).print()

    //处理完 业务逻辑后 手动提交offset维护到本地 mysql中
    // 实际生产频繁读写mysql性能不好，首选hbase或redis
    // 每一个批次都要写
    stream.foreachRDD(
      rdd => {
      val sqlProxy = new SqlProxy()
      val client = DataSourceUtil.getConnection
      try {
        //
        val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        for (or <- offsetRanges) {
          //用最新的offset更新mysql
          sqlProxy.executeUpdate(
            client,
            //sql语句
            "replace into `offset_manager` (groupid,topic,`partition`,untilOffset) values(?,?,?,?)",
            //把untilOffset封装进数组
            Array(groupid, or.topic, or.partition.toString, or.untilOffset)
          )
        }
      } catch {
        case e: Exception => e.printStackTrace()
      } finally {
        sqlProxy.shutdown(client)
      }
    })
    ssc.start()
    ssc.awaitTermination()
  }
}
